package com.gamejelly.proxy.client;

import com.gamejelly.proxy.pojo.RpcCall;
import com.gamejelly.proxy.pojo.RpcResponse;
import com.gamejelly.proxy.util.Timer;
import io.netty.channel.Channel;
import lombok.Getter;
import lombok.Setter;

import java.net.InetSocketAddress;
import java.util.concurrent.*;


public class Connection {

    //管道
    private Channel channel;

    //远程地址
    private final String remoteAddr;

    private CountDownLatch latch;

    private final ConcurrentHashMap<String, CompletableFuture<RpcResponse>> futures = new ConcurrentHashMap<>();

    private final ConnectionManager manager = ConnectionManager.getInstance();

    @Getter
    private volatile boolean active;

    @Setter
    private static int RETRY_NUMS = 2;

    public Connection(Channel channel) {
        this.channel = channel;
        this.active = channel.isActive();
        this.remoteAddr = ((InetSocketAddress) channel.remoteAddress()).getAddress().getHostAddress();
        this.latch = new CountDownLatch(1);
        channel.pipeline().get(ClientResponseHandler.class).setConnection(this);
    }

    public void reConnected(Channel channel) {
        this.channel = channel;
        this.active = channel.isActive();
        channel.pipeline().get(ClientResponseHandler.class).setConnection(this);
        this.latch = new CountDownLatch(1);
    }

    public void connected() {
        active = true;
        latch.countDown();
    }

    public void disConnected() {
        this.active = false;
        this.latch = new CountDownLatch(1);
        channel.close();
        manager.reConnected(this);
    }


    public void oneway(RpcCall msg) throws TimeoutException, InterruptedException {
        if (!latch.await(10, TimeUnit.SECONDS)) {
            throw new TimeoutException(remoteAddr + "not connected");
        } else
            channel.writeAndFlush(msg);
    }

    CompletableFuture<RpcResponse> call(RpcCall msg) throws InterruptedException, TimeoutException {
        return call(msg, RETRY_NUMS);
    }


    public CompletableFuture<RpcResponse> call(RpcCall msg, int retry) throws TimeoutException, InterruptedException {
        oneway(msg);
        String id = msg.getId();
        CompletableFuture<RpcResponse> future = null;
        if (retry == RETRY_NUMS) {
            future = new CompletableFuture<>();
            futures.put(id, future);
        }
        //单次请求超时
        Timer.INSTANCE.newTimeout(timeout -> {
            CompletableFuture<RpcResponse> f = futures.get(id);
            if (f != null) {
                if (retry <= 0) {
                    futures.remove(id);
                    f.completeExceptionally(new TimeoutException(remoteAddr + "time out" + msg.getClassName() + "." + msg.getMethodName()));
                } else {
                    //重试
                    call(msg, retry - 1);
                }
            }
        }, 5, TimeUnit.SECONDS);
        return future;
    }

    public void consume(RpcResponse response) {
        CompletableFuture<RpcResponse> future = futures.get(response.getRequestId());
        if (future != null) {
            future.complete(response);
        }
    }


    public void finish(RpcResponse response) {
        CompletableFuture<RpcResponse> future = futures.remove(response.getRequestId());
        if (future != null)
            future.complete(response);
    }

}
